package org.example.com.atguigu.day05;

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

import java.util.Properties;

public class ReaderFromMysql3 {
    public static void main(String[] args) {
        SparkSession spark = SparkSession.builder().master("local[4]").appName("test").getOrCreate();
        String url = "jdbc:mysql://hadoop102:3306/gmall?useUnicode=true&characterEncoding=UTF-8";
        String tableName = "user_info";
        Properties props = new Properties();
        props.setProperty("user", "root");
        props.setProperty("password", "000000");
        // todo 读取MySQL的第三种方式
        // TODO 此种方式生成的DataSet的分区数 = (upperBound - LowerBound) >= numPartitions ? numPartitions :  (upperBound - LowerBound);
        // TODO columName: 用于分区的mysql表的列名, 该列必须为数字,日期,时间戳类型 (一般使用mysql的主键)
        // TODO lowerBound: 用于计算分区步距的最小值 (一般设置columName列的最小值)
        // TODO upperBound: 用于计算分区步距的最大值 (一般设置columName列的最大值)

        // TODO 由于columnName列的最小/大值可能每天都会增加数据,变大;
        // TODO 所以我们去除columnNmae列的最小值和最大值,该数据集有且仅有一条数据
        Dataset<Row> ds1 = spark.read().jdbc(url, "(select min(id) min_id, max(id) max_id from user_info) user", props);
        Row row = ds1.first();
        Long min_id = (Long) row.getAs("min_id");
        Long max_id = (Long) row.getAs("max_id");
        System.out.println(min_id + " " + max_id);
        Dataset<Row> ds = spark.read().jdbc(url, tableName, "id", min_id, max_id, 5, props);
        // 查看分区数
        System.out.println(ds.javaRDD().getNumPartitions()); // 5 个
        // 保存数据 可以看到有5个文件
        ds.javaRDD().saveAsTextFile("output");
        // 展示数据
        ds.show();
    }
}
